package doyouknow.group.CnAndZhb.core;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import doyouknow.group.CnAndZhb.entity.TbcPoint;
import doyouknow.group.CnAndZhb.service.TbcPointService;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.SQLOutput;
import java.sql.Timestamp;
import java.util.ArrayList;
import doyouknow.group.CnAndZhb.utils.DateUtil;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author gaoyf
 * @since 2020/4/9 0009 16:05
 */
@Component
public class MQTTSubscriber extends MQTTConfig implements MqttCallback, IMQTTSubscriber {

    private String brokerUrl = null;
    final private String colon = ":";//冒号分隔符
    final private String clientId = "cn_sub_up";//客户端ID  这里可以随便定义

    private MqttClient mqttClient = null;
    private MqttConnectOptions connectionOptions = null;
    private MemoryPersistence persistence = null;
    @Resource
    private TbcPointService tbcPointService;


    private static final Logger logger = LoggerFactory.getLogger(MQTTSubscriber.class);

    public MQTTSubscriber() {
        this.config();
    }


    public void connectComplete(boolean b, String s) {
        System.out.println("丢失重连成功");
    }

    @Override
    public void connectionLost(Throwable cause) {
        logger.error("connect lost");
        int i = 3;
        while (true) {
            if (i < 0) {
                //这里是我自己为了失败3次后告警使用
                logger.error("2", "mqtt丢失问题");
                break;
            }
            if (!mqttClient.isConnected()) {
                System.out.println("***** client to connect *****");
                try {
                    //这个是30秒后重连
                    Thread.sleep(30000);
                    mqttClient.reconnect();
                } catch (Exception e) {
                    e.printStackTrace();
                    continue;
                }
                i--;
            }
            if (mqttClient.isConnected()) {
                System.out.println("***** connect success *****");
                break;
            }
        }
        //logger.error("连接丢失,原因{" + cause + "}");
        // 连接丢失后，一般在这里面进行重连
//        while (true) {
//            try {
//                Thread.sleep(30000);
//                this.mqttClient.reconnect();
////                this.mqttClient.subscribe("hotel_gateway");
//
//                logger.info("mqtt重新连接，重新订阅！");
//                break;
//            }catch (Exception e) {
//                e.printStackTrace();
//                continue;
//            }
//        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
//         Called when a message arrives from the server that matches any subscription made by the client
        String time = new Timestamp(System.currentTimeMillis()).toString();
        //System.out.println("消息到达时间：" + time + "  Topic: " + topic + "  Message: "
        //        + new String(message.getPayload()));


        String NewMessage = new String(message.getPayload());
        JSONObject jsonObj = JSONObject.parseObject(NewMessage);
        String Service = jsonObj.get("Service").toString();

        if (Service.equals("NotifyInfo")){
            String Name = jsonObj.get("Name").toString();
            if (Name.equals("OnOff")){
                System.out.println("***********************************************************************");
                System.out.println(jsonObj.toJSONString());
                String Mac = jsonObj.getString("Mac");
                String Point = jsonObj.getString("Point");
                String ReadTag = jsonObj.getString("ReadTag");
                String UpdateTime = jsonObj.getString("UpdateTime");

                TbcPoint insertPoint = new TbcPoint();
                insertPoint.setPointId("0");
                insertPoint.setPointNode("0");
                insertPoint.setUpdatetime(jsonObj.getString("UpdateTime"));
                insertPoint.setNetwork(jsonObj.getString("Network"));
                insertPoint.setGatewayMac(jsonObj.getString("GatewayMac"));
                insertPoint.setDeviceMac(jsonObj.getString("Mac"));
                insertPoint.setPointNumber(jsonObj.getString("Point"));
                insertPoint.setMonitorValue(jsonObj.getString("Value"));
                insertPoint.setReadtag(ReadTag==null?"inValid":ReadTag);

                TbcPoint tbcPointMacAndPoint = tbcPointService.queryByMacPoint(Mac, Point);

                    insertPoint.setPointId("0");
                    insertPoint.setPointNode("0");
                    insertPoint.setUpdatetime(jsonObj.getString("UpdateTime"));
                    insertPoint.setNetwork(jsonObj.getString("Network"));
                    insertPoint.setGatewayMac(jsonObj.getString("GatewayMac"));
                    insertPoint.setDeviceMac(jsonObj.getString("Mac"));
                    insertPoint.setPointNumber(jsonObj.getString("Point"));
                    insertPoint.setMonitorValue(jsonObj.getString("Value"));
                    insertPoint.setReadtag(ReadTag==null?"inValid":ReadTag);

                if (tbcPointMacAndPoint == null) {
                    System.out.println("记录不存在，增加数据 更新时间" + UpdateTime);

                    tbcPointService.insert(insertPoint);
                } else {
                    Integer id = tbcPointMacAndPoint.getId();
                    insertPoint.setId(id);
                    System.out.println(id + "存在，更新数据 更新时间" + UpdateTime );
                    tbcPointService.update(insertPoint);
                }
        }


            }
    }






    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Leave it blank for subscriber

    }

    @Override
    public void subscribeMessage(String topic) {
        try {
            this.mqttClient.subscribe(topic, this.qos);
        } catch (MqttException me) {
            me.printStackTrace();
        }
    }


    public void disconnect() {
        try {
            this.mqttClient.disconnect();
        } catch (MqttException me) {
            logger.error("ERROR", me);
        }
    }

    protected void config(String ip, Integer port, Boolean ssl, Boolean withUserNamePass) {
        String protocal = this.TCP;
        if (ssl) {
            protocal = this.SSL;
        }
        this.brokerUrl = protocal + this.ip + colon + port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();
        try {
            this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
            this.connectionOptions.setCleanSession(false);
            this.connectionOptions.setAutomaticReconnect(true);

            if (withUserNamePass) {
                if (password != null) {
                    this.connectionOptions.setPassword(this.password.toCharArray());
                }
                if (username != null) {
                    this.connectionOptions.setUserName(this.username);
                }
            }
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

    protected void config() {

        this.brokerUrl = this.TCP + this.ip + colon + this.port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();
        try {
            this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
            this.connectionOptions.setCleanSession(false);
            this.connectionOptions.setAutomaticReconnect(true);
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

}
